package lihua.robot.utils;

import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @description: 功能描述
 * @author: XuRuoHan
 * @date: 2024-04-03 21:50
 */
@Slf4j
@Component
public class MqttServer {

    private MqttClient client;
    private String[] topics;

    final String url = "tcp://172.16.1.100:1883";
    final String clientId = "7d7f95f616204c57af9ba84db3ed3a53";
    final int qos = 2;
    final String username = "";
    final String password = "";
    /**
     * 每秒处理300条消息
     */
    private final RateLimiter rateLimiter = RateLimiter.create(100);

    /**
     * 手动创建一个 WorkStealingPool，线程数为 Runtime.getRuntime().availableProcessors(),CachedThreadPool 或 WorkStealingPool
     */
    private static final ExecutorService THREAD_POOL = Executors.newWorkStealingPool(Runtime.getRuntime().availableProcessors() / 4);

    @PostConstruct
    public void execute() {
        try {
            topics = new String[]{
                    "/sys/WG581WLAN22101302145/up",
                    "/sys/WG581WLAN22101302154/up",
                    "/sys/WG581WLAN22101302155/up",
                    "/sys/WG581WLAN22101302141/up",
                    "/sys/WG581WLAN22101302118/up",
                    "/sys/WG581WLAN22101302119/up",
                    "/sys/WG581WLAN22101701119/up",
                    "/sys/WG581WLAN22101302138/up"
            };
//            topics = Collections.singletonList("/sys/WG581WLAN22101302154/up").toArray(new String[0]);
            client = new MqttClient(url, clientId);
            reconnectAndSubscribe();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void reconnectAndSubscribe() {
        try {
            MqttCallback mqttCallback = new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
//                    System.out.println("【连接丢失】");
//                    System.out.println(throwable.getMessage());
                    reconnectAndSubscribe();
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 消息处理前请求许可
                    if (rateLimiter.tryAcquire()) {
                        // 处理消息
                        try {
                            THREAD_POOL.execute(() -> {
                                try {
//                                    System.out.println("【信息到达】：" + topic);
//                                    System.out.println(message);
                                    String jsonPayload = new String(message.getPayload());

                                    //解析json
                                    JSONObject jsonObject = JSONUtil.parseObj(jsonPayload);
                                    String now_devlist = jsonObject.getStr("devList");
                                    String now_devList = now_devlist.replace("[", "").replace("]", "");

                                    JSONObject devList = JSONUtil.parseObj(now_devList);
                                    String now_varList = devList.getStr("varList");
                                    JSONObject varList =  JSONUtil.parseObj(now_varList);

                                    Integer dqcd = varList.getInt("dqcd");//当前长度
                                    Integer devSn = devList.getInt("devSn");//设备号
                                    Integer cmdId = jsonObject.getInt("cmdId");



                                    //存入redis
                                    if(cmdId == 103 && dqcd != null) {
                                        System.out.println(DateUtil.date() +"----设备"+devSn+":"+dqcd);
                                        RedisUtil.storeDataInCache(devSn,dqcd);
                                    }
                                } catch (Exception e) {

                                }
                            });

                        } catch (Exception e) {
                            System.out.println("【message异常】");
                            System.out.println(e.getMessage());
                        }
                    } else {
                        System.out.println("【丢弃topic】" + topic);
                        System.out.println(message);
                    }
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("【deliveryComplete】=========================");
                }
            };
            // 订阅mqtt消息
            this.subscribes(url, clientId, qos, username, password, topics, mqttCallback);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void subscribes(String brokerUrl, String clientId, int qqs, String username, String password, String[] topics, MqttCallback callback) throws MqttException {

        try {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setAutomaticReconnect(true);
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setMaxInflight(10);
            options.setKeepAliveInterval(60);
            options.setConnectionTimeout(30);
            client.connect(options);
            client.setCallback(callback);
            client.subscribe(topics);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
